package com.github.jiangxch.rpc.transport.exchanger.jdk;

import com.github.jiangxch.rpc.transport.Channel;
import com.github.jiangxch.rpc.transport.ChannelHandler;
import com.github.jiangxch.rpc.transport.exchanger.ExchangeChannel;
import com.github.jiangxch.rpc.transport.exchanger.ExchangeChannelHandler;
import com.github.jiangxch.rpc.transport.exchanger.Replier;
import com.github.jiangxch.rpc.transport.exchanger.Request;
import com.github.jiangxch.rpc.transport.exchanger.Response;

import java.io.IOException;
import java.util.concurrent.CompletableFuture;

/**
 * @author: jiangxch
 * @date: 2021/2/26 上午1:22
 */
public class JdkExchangeChannelHandler implements ExchangeChannelHandler {
    // 上层业务的 channelHandler
    private Replier replier;
    
    public JdkExchangeChannelHandler(Replier replier) {
        this.replier = replier;
    }
    
    @Override
    public void onSend(Channel channel, Object message) {
    }
    
    @Override
    public void onActive(Channel channel) throws IOException {
    }
    
    @Override
    public void onReceive(Channel channel, Object message) throws IOException {
        if (message instanceof Request) {
            handlerRequest(new JdkExchangeChannel(channel),(Request)message);
        } else if (message instanceof Response) {
            handlerResponse((Response)message);
        } else {
            throw new RuntimeException("cna't handler message,message="+message.toString());
        }
    }
    
    private void handlerResponse(Response message) {
        FutureManager.receiveResponse(message);
    }
    
    private void handlerRequest(ExchangeChannel channel,Request message) {
        CompletableFuture<Object> future = CompletableFuture.supplyAsync(() -> {
            return replier.replay(channel, message.getData());
        });
        future.thenAccept((replayResult) -> {
            Response response = new Response();
            response.setRequestId(message.getRequestId());
            response.setStatus(Response.ok);
            response.setResult(replayResult);
            try {
                channel.send(response);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        });
    }
    
    @Override
    public void caught(Channel channel, Throwable throwable) {
    
    }
}
